package com.ruoyi.common.utils;

import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.function.Consumer;
import java.util.function.Function;

import static com.ruoyi.common.config.AsyncConfig.*;

/**
 * @author chenhainan
 * @since 2024-05-15 17:06
 */
@Slf4j
public class AsyncUtils {

    private AsyncUtils() {
    }

    /**
     * 异步消费 list内容
     *
     * @param list     数据
     * @param consumer 转换器
     * @param <T>      输入类型
     */
    public static <T> void consumeAsync(List<T> list, Consumer<T> consumer) {
        if (Objects.isNull(list) || list.isEmpty()) {
            return;
        }
        if (list.size() == 1) {
            consumer.accept(list.getFirst());
        }
        final CountDownLatch countDownLatch = new CountDownLatch(list.size());
        for (T t : list) {
            Thread.startVirtualThread(() -> {
                try {
                    consumer.accept(t);
                } catch (Exception e) {
                    log.error("消费[{}]发生错误", t, e);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            log.error("异步消费[{}]发生异常", list, e);
        }
    }

    /**
     * 等待所有线程任务完成
     *
     * @param threads 线程任务
     */
    public static void joinAll(List<Thread> threads) {
        for (Thread thread : threads) {
            try {
                thread.join();
            } catch (InterruptedException e) {
                log.error("thread [{}] interrupted", thread.getName());
            }
        }
    }


    /**
     * 异步转换 list内容为新的内容
     *
     * @param list   数据
     * @param mapper 转换器
     * @param <T>    输入类型
     * @param <R>    输出类型
     * @return 新的list
     */
    public static <T, R> List<R> mapAsync(List<T> list, Function<T, R> mapper) {
        if (Objects.isNull(list) || list.isEmpty()) {
            return Collections.emptyList();
        }
        if (list.size() == 1) {
            return List.<R>of(mapper.apply(list.getFirst()));
        }
        return getAll(list.stream().map(t -> VIRTUAL_EXECUTOR.submit(() -> mapper.apply(t))).toList());
    }

    @SafeVarargs
    public static <T> List<T> getAll(Future<T>... futures) {
        return getAll(Arrays.asList(futures));
    }


    /**
     * 等待所有 future 完成
     *
     * @param futures future 列表
     * @param <T>     类型
     * @return 所有的 future 结果
     */
    public static <T> List<T> getAll(List<Future<T>> futures) {
        final var resultList = new ArrayList<T>(futures.size() + 1);
        for (Future<T> future : futures) {
            try {
                resultList.add(future.get());
            } catch (InterruptedException | ExecutionException e) {
                log.error("future interrupted", e);
            }
        }
        return resultList;
    }

    public static void main(String[] args) {
        final long start = System.currentTimeMillis();
        consumeAsync(List.of(1, 2, 3, 4, 5, 6), i -> {
            try {
                System.out.println(Thread.currentThread().getName());
                Thread.sleep(1000L);
                System.out.println(i);
            } catch (InterruptedException e) {
                log.error("", e);
            }
        });
        final long end = System.currentTimeMillis();
        System.out.println(end - start);
    }
}
